from re import compile
from typing import Any

from fastapi import BackgroundTasks
from sqlalchemy.exc import NoResultFound
from sqlalchemy.sql.functions import count
from sqlmodel import or_, select
from sqlmodel.ext.asyncio.session import AsyncSession

from app.core.config import SETTING
from app.models.sql import BookChapter
from app.models.sql.Book import Book
from app.schema.book import BookCatalogItemResponseModel
from app.services.cache_service import cache, cache_get, cache_set
from app.utils import Reverse
from app.utils.codec import PydanticCodec, PydanticListCodec

key_exclude_pattern = compile(r"[^a-z0-9\u4e00-\u9fa5\s]")


# --- 工具函数 ---
def escape_like_pattern(value: str) -> str:
    """
    转义 SQL LIKE 中的特殊字符：% _ \
    """
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def normalize_search_keyword(keyword: str, max_length: int = 50) -> str:
    """
    标准化搜索关键字：
    - 去除首尾空白
    - 转为小写（可选，根据业务需求）
    - 限制长度
    - 只保留字母、数字、中文、空格（防止特殊字符污染缓存 key）
    """
    if not keyword or not isinstance(keyword, str):
        return ""

    # 限制长度
    if len(keyword) > max_length:
        keyword = keyword[:max_length]

    # 去除首尾空白
    keyword = keyword.strip()
    if not keyword:
        return ""

    # 可选：转小写（提升缓存命中率）
    keyword = keyword.lower()

    # 只保留安全字符：字母、数字、中文、空格
    # 根据业务需求调整正则（如需支持英文标点可扩展）
    keyword = key_exclude_pattern.sub("", keyword)

    # 合并多个空格为一个，并去除首尾空格
    keyword = " ".join(keyword.split())

    return keyword


class BookService:

    @staticmethod
    @cache(expire=SETTING.BOOK_CACHE_EXPIRE, exclude_kwargs=["database"])
    async def get_category(
            database: AsyncSession
    ) -> list[str]:
        """
        获取图书分类
        return: list[str]分类列表
        """
        statement = select(Book.category).distinct()
        result = await database.exec(statement)
        categories = result.all()
        return [str(category) for category in categories]

    @staticmethod
    @cache(
        expire=SETTING.BOOK_CACHE_EXPIRE,
        exclude_kwargs=["database"],
        key_prefix="get_book_by_id",
        codec=PydanticCodec(Book),
    )
    async def get_book_by_id(
            book_id: int,
            database: AsyncSession
    ) -> dict[str, Any]:
        """
        获取图书信息
        :param book_id: 图书ID
        :param database:    数据库会话
        :return:    图书信息
        """
        try:
            statement = select(Book).where(Book.id == book_id)
            result = await database.exec(statement)
            book = result.one_or_none()
            book.cover = f"{SETTING.SERVER_URL}/static/book/{book.id}/{book.cover}"
            return book
        except NoResultFound:
            raise ValueError("图书不存在")

    @staticmethod
    async def get_book_by_list(
            book_ids: list[int],
            database: AsyncSession,
            background_tasks: BackgroundTasks
    ) -> list[Book]:
        """
        获取图书信息

        :param book_ids:  图书ID列表
        :param database:        数据库会话
        :param background_tasks: 后台任务
        :return:         图书信息
        """
        if not book_ids:
            return []
        miss_book_ids = []
        book_list = []
        for book_id in book_ids:
            try:
                result = await cache_get(
                    args=[],
                    kwargs={"book_id": book_id},
                    key_prefix="get_book_by_id",
                    codec=PydanticCodec(Book),
                )
                if result:
                    book_list.append(result)
                else:
                    miss_book_ids.append(book_id)
            except Exception as e:
                raise ValueError(f"获取图书信息失败{str(e)}")
        if miss_book_ids:
            statement = select(Book).where(Book.id.in_(miss_book_ids))
            result = await database.exec(statement)
            books = result.all()
            for book in books:
                book.cover = f"{SETTING.SERVER_URL}/static/book/{book.id}/{book.cover}"
            _ = [background_tasks.add_task(
                cache_set,
                args=[],
                kwargs={"book_id": book.id},
                key_prefix="get_book_by_id",
                value=book,
                codec=PydanticCodec(Book)
            ) for book in books
            ]
            book_list.extend(books)
        return book_list

    @staticmethod
    @cache(
        expire=SETTING.BOOK_CACHE_EXPIRE,
        exclude_kwargs=["database"],
        codec=PydanticListCodec(BookCatalogItemResponseModel),
    )
    async def get_book_toc_by_id(
            book_id: int,
            database: AsyncSession
    ) -> list[BookCatalogItemResponseModel]:
        """
        获取图书目录
        :param book_id: 图书ID
        :param database:      数据库会话
        :return:       图书目录
        """
        statement = select(BookChapter.id, BookChapter.title).where(BookChapter.book_id == book_id)
        result = await database.exec(statement)
        toc = result.all()
        return [BookCatalogItemResponseModel(id=chapter_id, title=title) for chapter_id, title in toc]

    @staticmethod
    @cache(expire=SETTING.BOOK_CACHE_EXPIRE, exclude_kwargs=["database"])
    async def get_book_chapter_by_id(
            chapter_id: int,
            database: AsyncSession
    ) -> str:
        """
        获取图书章节内容

        :param chapter_id:    章节ID
        :param database:         数据库会话
        :return:          章节内容
        """
        statement = select(BookChapter.content).where(BookChapter.id == chapter_id)
        result = await database.exec(statement)
        chapter = result.one_or_none()
        return str(chapter)

    @staticmethod
    @cache(expire=SETTING.BOOK_CACHE_EXPIRE, exclude_kwargs=["database"])
    async def get_book_chapter_by_index(
            book_id: int,
            chapter_index: int,
            database: AsyncSession,
    ) -> str:
        """
        获取图书章节内容
        :param book_id:        图书ID
        :param chapter_index:     章节索引
        :param database:             数据库会话
        :return:             章节内容
        """
        statement = select(BookChapter.content) \
            .where(BookChapter.book_id == book_id) \
            .order_by(BookChapter.sort_order) \
            .limit(1) \
            .offset(chapter_index)
        result = await database.exec(statement)
        chapter = result.one_or_none()
        return str(chapter)

    @staticmethod
    @cache(expire=SETTING.BOOK_CACHE_EXPIRE, exclude_kwargs=["database"])
    async def get_books_total_count(
            database: AsyncSession,
    ) -> int:
        """
        获取图书总数
        :param database:      数据库会话
        :return:              图书总数
        """
        statement = select(count(Book.id))
        result = await database.exec(statement)
        return result.one_or_none()

    @staticmethod
    @cache(
        expire=SETTING.BOOK_CACHE_EXPIRE,
        exclude_kwargs=["database"],
        codec=PydanticListCodec(Book),
    )
    async def search_book(
            *,
            keyword: str,
            database: AsyncSession,
    ) -> list[Book]:
        """
        搜索图书
        :param keyword:   搜索关键字
        :param database:      数据库会话
        :return:              图书列表
        """
        normalized_keyword = normalize_search_keyword(keyword)
        if not normalized_keyword:
            return []
        # 2. 转义 LIKE 通配符，防止意外匹配或注入
        escaped_keyword = escape_like_pattern(normalized_keyword)
        pattern = f"{escaped_keyword}%"

        # 3. 使用参数化查询（SQLAlchemy 自动处理参数绑定）
        statement = select(Book).where(
            or_(
                Book.name.like(pattern, escape="\\"),
                Book.author.like(pattern, escape="\\")
            )
        )

        # 4. 执行查询
        result = await database.exec(statement)
        books = result.all()
        for book in books:
            book.cover = f"{SETTING.SERVER_URL}/static/book/{book.id}/{book.cover}"
        # 5. 返回序列化结果
        return books

    @staticmethod
    @cache(
        expire=SETTING.BOOK_CACHE_EXPIRE,
        exclude_kwargs=["database"],
        codec=PydanticListCodec(Book)
    )
    async def get_books_by_category(
            *,
            database: AsyncSession,
            category: str,
    ) -> list[Book]:
        """
        获取图书列表
        :param database:      数据库会话
        :param category:         分类
        :return:               图书列表
        """
        statement = select(Book).where(Book.category == category)
        result = await database.exec(statement)
        books = result.all()
        for book in books:
            book.cover = f"{SETTING.SERVER_URL}/static/book/{book.id}/{book.cover}"
        return books

    @staticmethod
    async def get_book_select(
            *,
            database: AsyncSession,
            limit: int,
            offset: int,
            category: str,
            sorted_condition: dict[str, int],
    ) -> list[Book]:
        """
        获取图书列表
        :param database:      数据库会话
        :param limit:           每页数量
        :param offset:           偏移量
        :param category:         分类
        :param sorted_condition 排序条件
        :return:               图书列表
        """
        books = await BookService.get_books_by_category(database=database, category=category)

        def sort_key(book):
            key_parts = []
            for field, order in sorted_condition.items():
                val = book.get(field)

                # 统一处理 None：我们希望 None 始终排在最后
                if val is None:
                    # 用一个极大对象表示 "最后"
                    val = (1, None)  # (优先级=1, 值)
                else:
                    val = (0, val)  # (优先级=0, 值)

                # 如果是降序（-1），用 Reverse 包装整个 (0, val)
                if order == -1:
                    val = Reverse(val)

                key_parts.append(val)

            return tuple(key_parts)

        sorted_books = sorted(books, key=sort_key)
        return sorted_books[offset: offset + limit]

    @staticmethod
    @cache(expire=SETTING.BOOK_CACHE_EXPIRE, exclude_kwargs=["database"])
    async def get_category_count(
            *,
            database: AsyncSession,
            category: str,
    ):
        """
        获取图书数量
        :param database:      数据库会话
        :param category:         分类
        :return:               图书数量
        """
        statement = select(count(Book.id)).where(Book.category == category)
        result = await database.exec(statement)
        book_ids = result.one_or_none()
        return book_ids if book_ids else 0
